package com.yanqu.road.connect;

import com.yanqu.road.message.YanQuMessage;
import com.yanqu.road.task.FlushConnectionTask;
import com.yanqu.road.utils.YanQuMessageUtils;
import io.netty.channel.Channel;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class YanQuConnection extends Connection {

	private static final int WRITE_LIMIT_MIN = 64;
	private static final int WRITE_LIMIT_MAX = 128;

	private static Logger logger = LogManager.getLogger(YanQuConnection.class.getName());


	private int linkIndex;

	public int getLinkIndex() {
		return linkIndex;
	}

	private Channel channel;

	public Channel getChannel() {
		return channel;
	}

	private boolean canWrite;

	private boolean freshAndGetCanWrite() {
		if (canWrite) {
			if (counter.get() >= WRITE_LIMIT_MAX) {
				canWrite = false;
			}
		} else {
			if (counter.get() <= WRITE_LIMIT_MIN) {
				canWrite = true;
			}
		}
		return canWrite;
	}

	public YanQuConnection(Channel channel, boolean isValidate, int queueSize, int linkIndex) {
		super(isValidate);
		flushSessionTask = new FlushConnectionTask(this);
		flushSessionTaskSwitch = new AtomicBoolean(true);
		this.channel = channel;
		this.linkIndex = linkIndex;
		messageQueue = new ArrayBlockingQueue<>(queueSize);
		counter = new AtomicInteger();
		listener = arg0 -> {
			counter.decrementAndGet();
			if (freshAndGetCanWrite()) {
				flush();
			}
		};

	}



	private Runnable flushSessionTask;

	public AtomicInteger getCounter() {
		return counter;
	}


	private AtomicBoolean flushSessionTaskSwitch;




	public AtomicBoolean getFlushSessionTaskSwitch() {
		return flushSessionTaskSwitch;
	}



	private BlockingQueue<YanQuMessage> messageQueue;

	public BlockingQueue<YanQuMessage> getMessageQueue() {
		return messageQueue;
	}

	private AtomicInteger counter;

	public void send(int playerId, YanQuMessage msg) {
		msg.setPlayerId(playerId);
		write(msg);
	}


	private GenericFutureListener<Future<? super Void>> listener;

	public GenericFutureListener<Future<? super Void>> getListener() {
		return listener;
	}



	public void send(YanQuMessage msg) {
		write(msg);
	}



	@Override
	public void write(YanQuMessage msg) {
		if (msg == null) {
			return;
		}
		try {
            if (YanQuMessageUtils.isLimitPacketSize(msg)) {
                logger.error("packet size bigger than 50M, refuse message code {}, player {}", msg.getCode(), msg.getPlayerId());
//			return;
            }
        }catch (Exception e){
			logger.error("isLimitPacketSize error:", e);
        }
		boolean oResult = messageQueue.offer(msg);
		if(!oResult){
			logger.error("offer messageQueue fail,code:{},userId:{}", msg.getCode(), msg.getPlayerId());
		}
//        try {
//            messageQueue.put(msg);
//        } catch (InterruptedException e) {
//			logger.error("put messageQueue error,code:{},userId:{},:", msg.getCode(), msg.getPlayerId(), e);
//        }
        if (freshAndGetCanWrite()) {
			flush();
		}
	}

	@Override
	public boolean isAlive() {
		try {
			if (channel == null) {
				return false;
			} else {
				return channel.isActive();
			}
		} catch (Exception e) {
			logger.error("channel is not active,error:", e);
			return false;
		}
	}

	@Override
	public void flush() {
		if (messageQueue.isEmpty()) {
			return;
		}

		if (flushSessionTaskSwitch.getAndSet(false)) {
			EventExecutor executor = channel.eventLoop();
			executor.execute(flushSessionTask);
		}
	}

	@Override
	public void close() {
		channel.close();
	}


}
